/** @file $RCSfile: tbusd.c,v $
  general description of this module
  $Id: tbusd.c,v 1.1.1.1 2013/05/27 02:58:11 ouyang Exp $
@author $Author: ouyang $
@date $Date: 2013/05/27 02:58:11 $
@version $Revision: 1.1.1.1 $
@note Editor: Vim 6.3, Gcc 4.0.2, tab=4
@note Platform: Linux
*/


#include <assert.h>
#include <string.h>
#include <errno.h>


#include "comm/tmempool.h"
#include "tdr/tdr.h"
#include "pal/pal.h"
#include "tapp/tapp.h"
#include "tbus/tbus_macros.h"
#include "tbus/tbus_route.h"
#include "tbusd_misc.h"
#include "tbusd_log.h"
#include "tbus/tbus.h"
#include "tbus/tbus_channel.h"



#define RELAY_CONFINFO "TbusRelay"

#define RELAY_VER_MAJOR 1
#define RELAY_VER_MINOR 0
#define RELAY_VER_REV    2
#define RELAY_VER_BUILD  20090713

extern unsigned char g_szMetalib_tbusd[] ;


RELAYENV gs_stRelayEnv;
static TAPPCTX gs_stAppCtx;


/**
  various internal function defined
*/
static int tbusd_init(TAPPCTX *a_pstAppCtx,  LPRELAYENV a_pstRunEnv);


static void tbusd_init_stat(LPRELAYENV  a_pstRunEnv);

int app_epoll_create( LPRELAYENV  a_pstRunEnv ) ;
int	app_listen_create(LPRELAYENV  a_pstRunEnv) ;


/** 执行程序退出时的清除操作
*/
static int tbusd_fini(IN TAPPCTX *a_pstAppCtx,  IN LPRELAYENV a_pstRunEnv);

/** tapp 时钟周期触发时执行的逻辑：输出统计信息，检查relay的链接尝试重连
*/
static int tbusd_tick(IN TAPPCTX *a_pstAppCtx,  IN LPRELAYENV a_pstRunEnv);

/** 当tbusd idle时将epoll_wait的时间改成 10毫秒等待
*/
static int tbusd_idle(IN TAPPCTX *a_pstAppCtx,  IN LPRELAYENV a_pstRunEnv);

int relay_channel_attach ( LPRELAYENV  a_pstRunEnv ) ;


/** tbusd主循环逻辑
*/
int tbusd_proc(IN TAPPCTX *a_pstAppCtx,  IN LPRELAYENV a_pstRunEnv);

int pkg_sent_loop ( IN const int a_iCnt ) ;
int pkg_recv_sent ( IN const int a_iSocket, INOUT RELAY *a_ptRelay ) ;
int pkg_sent_recv ( IN const int a_iSocket, INOUT RELAY *a_ptRelay ) ;

int TbusdServiceInit( int argc, char ** argv )
{
	int iRet;
	void *pvArg = &gs_stRelayEnv;

	//printf("sizeof TBUSSHMGCIM = %d\n", sizeof(TBUSSHMGCIM));
	//printf("sizeof TBUSSHMGRM = %d\n", sizeof(TBUSSHMGRM));

	memset(&gs_stRelayEnv, 0 , sizeof(gs_stRelayEnv));
	memset(&gs_stAppCtx, 0, sizeof(gs_stAppCtx));

	gs_stAppCtx.argc	=	argc;
	gs_stAppCtx.argv	=	argv;

	gs_stAppCtx.pfnInit	=	(PFNTAPPFUNC)tbusd_init;
	gs_stAppCtx.pfnFini	=	(PFNTAPPFUNC)tbusd_fini;
	gs_stAppCtx.pfnProc	=	(PFNTAPPFUNC)tbusd_proc;
	gs_stAppCtx.pfnTick	=	(PFNTAPPFUNC)tbusd_tick;
	gs_stAppCtx.pfnIdle	=	(PFNTAPPFUNC)tbusd_idle;


	/*初始化系统数据元数据库*/
	gs_stAppCtx.iLib	=	(unsigned long int)&g_szMetalib_tbusd[0];
	gs_stAppCtx.stConfData.pszMetaName	=	RELAY_CONFIG_DATA_META_NAME;
	gs_stAppCtx.stRunDataStatus.pszMetaName = RELAY_RUN_DATA_META_NAME;
	gs_stAppCtx.uiVersion = TAPP_MAKE_VERSION(RELAY_VER_MAJOR, RELAY_VER_MINOR, RELAY_VER_REV, RELAY_VER_BUILD);

	iRet = tapp_def_init(&gs_stAppCtx, pvArg);

	return iRet;
}

int TbusdService()
{
	int iRet;
	void *pvArg = &gs_stRelayEnv;

	iRet = tapp_def_mainloop(&gs_stAppCtx, pvArg);

	return iRet;
}

void TbusdServiceStop( )
{
	tapp_exit_mainloop();
}

#ifndef WIN32
int main ( int argc, char ** argv )
{
	int iRet = 0;
	void *pvArg = &gs_stRelayEnv;

	iRet = TbusdServiceInit(argc, argv);
	if( iRet<0 )
	{
		printf("Error: tapp_def_init failed, ret: %d\n", iRet);
		return iRet;
	}

	iRet = TbusdService();

	TbusdServiceStop();

	return iRet;
}
#endif


int tbusd_init(TAPPCTX *a_pstAppCtx,  LPRELAYENV a_pstRunEnv)
{
	int iRet = TBUS_SUCCESS ;
	time_t tNow;
	LPTBUSD pstConf ;

	assert(NULL != a_pstAppCtx);
	assert(NULL != a_pstRunEnv);

	/*检查tapp对 配置和运行数据结构的初始化结果 */
	a_pstRunEnv->pstConf = (LPTBUSD)a_pstAppCtx->stConfData.pszBuff;
	pstConf = a_pstRunEnv->pstConf;
	if ((NULL == a_pstRunEnv->pstConf) || ((int)sizeof(TBUSD) > a_pstAppCtx->stConfData.iLen)||
		(NULL == (LPTDRMETA)a_pstAppCtx->stConfData.iMeta))
	{
		printf("ERROR:tbusd_init, NULL config data or invalid size<%d>",
			a_pstAppCtx->stConfData.iLen) ;
		return TBUS_ERROR ;
	}
	a_pstRunEnv->pstStat = (LPTBUSDSTAT)a_pstAppCtx->stRunDataStatus.pszBuff;
	if ((NULL == a_pstRunEnv->pstStat) || ((int)sizeof(TBUSDSTAT) > a_pstAppCtx->stRunDataStatus.iLen)||
		(NULL == (LPTDRMETA)a_pstAppCtx->stRunDataStatus.iMeta))
	{
		printf("ERROR:tbusd_init, NULL run data or invalid size<%d>, or NULL meta",
			a_pstAppCtx->stRunDataStatus.iLen) ;
		return TBUS_ERROR ;
	}
    if (RELAY_DEFAULT_PKG_MAX_SIZE > pstConf->iPkgMaxSize)
    {
		pstConf->iPkgMaxSize = RELAY_DEFAULT_PKG_MAX_SIZE;
    }

	/* init debug log */
	iRet = tbusd_log_init(a_pstAppCtx, a_pstRunEnv);
	if (0 != iRet)
	{
		printf("failed to init log system ");
		return iRet;
	}

	tlog_log(g_ptRelayLog, TBUSD_PRIORITY_FATAL,0,0,  "CALL: relay module initialize..." ) ;
	tbusd_log_data(g_ptRelayLog, TBUSD_PRIORITY_FATAL, (LPTDRMETA)a_pstAppCtx->stConfData.iMeta,
		pstConf, sizeof(RELAYENV));


	/*init tbus api*/
	tbus_register_bussid(pstConf->iBussinessID);
	iRet = tbus_init_ex(pstConf->szGCIMShmKey, 0);
	if (TBUS_SUCCESS != iRet)
	{
		tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_FATAL, "tbus_init_ex failed, iRet %x", iRet);
		return iRet;
	}

	/* attach GCIM and GRM in share memory */
	iRet = tbus_get_gcimshm(&a_pstRunEnv->pstShmGCIM, pstConf->szGCIMShmKey, pstConf->iBussinessID,	0,
		0);
	if ( 0 != iRet )
	{
		tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_ERROR, "get GCIM failed  by shmkey<%s> and bussinessid %d, iRet %x",
			pstConf->szGCIMShmKey, pstConf->iBussinessID, iRet) ;
	}else
	{
		iRet = tbus_get_grmshm(&a_pstRunEnv->pstShmGRM, pstConf->szRelayShmKey, pstConf->iBussinessID,0);
		if (0 != iRet)
		{
			tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_FATAL,  "failed to get GRM tables by key %s and businessid %d, iRet %x",
				pstConf->szRelayShmKey, pstConf->iBussinessID, iRet) ;
		}
	}


	/*attach ralay channel*/
	if (0 == iRet)
	{
		iRet = relay_channel_attach(a_pstRunEnv) ;
	}

	/* init mem node pool */
	if (0 == iRet)
	{
		iRet = tmempool_new ( &a_pstRunEnv->ptMemPool, RELAY_MAX_EVENTS, sizeof(CONNNODE) ) ;
		if ( 0 > iRet )
		{
			tbusd_log_msg(g_ptRelayLog, TBUSD_PRIORITY_ERROR, "initialize, tmempool_new() failed %i\n", iRet ) ;
		}
	}

	/*init epoll*/
	if (0 == iRet)
	{
		iRet = app_epoll_create(a_pstRunEnv) ;
	}

	/*create listen socket*/
	if (0 == iRet )
	{
		iRet = app_listen_create(a_pstRunEnv) ;
		if ( TBUS_SUCCESS != iRet )
		{
			fprintf ( stderr, "FATAL: Listen at '%s' faild, process aborted...\n", a_pstRunEnv->pstConf->szListen ) ;
		}
	}

	if (0 == iRet)
	{
		/*init stat info*/
		LPTBUSSHMGRM pstShmGRMBackup;
		tbusd_init_stat(a_pstRunEnv);

		time(&tNow);
		tbusd_conncet_peer(a_pstRunEnv, tNow);

		a_pstRunEnv->bNeedRefresh = TDR_FALSE;
		a_pstRunEnv->dwGCIMVersion = a_pstRunEnv->pstShmGCIM->stHead.dwVersion;
		pstShmGRMBackup = &a_pstRunEnv->stShmGRMBackup;
		memcpy(&pstShmGRMBackup->stHead, &a_pstRunEnv->pstShmGRM->stHead, sizeof(pstShmGRMBackup->stHead));
		memcpy(&pstShmGRMBackup->astRelays[0], &a_pstRunEnv->pstShmGRM->astRelays[0],
				sizeof(TBUSSHMRELAYCNF)*(a_pstRunEnv->pstShmGRM->stHead.dwUsedCnt));


		a_pstRunEnv->tLastStat = tNow;
		a_pstRunEnv->tLastCheckHandShake = tNow;
		a_pstRunEnv->tLastRefreshConf = tNow;

		if (0 == a_pstAppCtx->iEpollWait)
		{
			a_pstAppCtx->iEpollWait = TAPP_EPOLL_WAIT;
		}
		a_pstRunEnv->iEpollWaitTime = a_pstAppCtx->iEpollWait;
		a_pstRunEnv->iMaxSendPkgPerLoop = pstConf->iMaxSendPerloop;

		tlog_log(g_ptRelayLog, TBUSD_PRIORITY_FATAL,0,0,  "tbusd initialize...ok" ) ;
	}/*if (0 == iRet)*/


	return iRet ;
}

void tbusd_init_stat(LPRELAYENV a_pstRunEnv)
{
	LPTBUSDSTAT pstStat;


	assert(NULL != a_pstRunEnv);
	assert(NULL != a_pstRunEnv->pstConf);
	assert(NULL != a_pstRunEnv->pstStat);


	pstStat = a_pstRunEnv->pstStat;
	memset(pstStat, 0, offsetof(TBUSDSTAT, astRelayConnStat));
	pstStat->dwRelayCnt = a_pstRunEnv->dwRelayCnt;
	memset(&pstStat->astRelayConnStat[0], 0, sizeof(TBUSRELAYCONNSTAT)*pstStat->dwRelayCnt);

}

/** 执行程序退出时的清除操作
*/
int tbusd_fini(TAPPCTX *a_pstAppCtx,  LPRELAYENV a_pstRunEnv)
{
	LPCONNNODE pstConn;
	unsigned int i;
	assert(NULL != a_pstAppCtx);

	assert(NULL != a_pstRunEnv);

	tlog_log(g_ptRelayLog, TBUSD_PRIORITY_FATAL,0,0,  "tbusd_fini, server quit..." ) ;


	if (NULL != a_pstRunEnv->ptMemPool)
	{
		int idx, pos = -1;
		idx = tmempool_find_used_first(a_pstRunEnv->ptMemPool, &pos);
		if (0 <= idx)
		{
			while ((idx = tmempool_find_used_next(a_pstRunEnv->ptMemPool, &pos)) >= 0)
			{   /*关闭所有的连接*/
				pstConn = (LPCONNNODE)tmempool_get(a_pstRunEnv->ptMemPool, idx);
				assert(NULL != pstConn);
				tmempool_find_used_next(a_pstRunEnv->ptMemPool, &idx);
				tbusd_destroy_conn(a_pstRunEnv, pstConn);
			}
		}

		tmempool_destroy(&a_pstRunEnv->ptMemPool);
	}/*if (NULL != a_pstRunEnv->ptMemPool)*/

	if (0 <= a_pstRunEnv->iEpoolFd)
	{
		epoll_destroy(a_pstRunEnv->iEpoolFd);
		a_pstRunEnv->iEpoolFd = -1;
	}

	/*释放relay管理通道*/
	for (i = 0; i < a_pstRunEnv->dwRelayCnt; i++)
	{
		LPRELAY pstRelay = a_pstRunEnv->pastTbusRelay[i];
		if (NULL != pstRelay)
		{
			free(pstRelay);
			a_pstRunEnv->pastTbusRelay[i] = NULL;
		}
	}/*for (i = 0; i < a_pstRunEnv->stRelayGlobal.dwRelayCnt; i++)*/
	a_pstRunEnv->dwRelayCnt = 0;

	tbusd_log_fini();

	return 0;
}

int relay_enable (LPTBUSSHMRELAYCNF a_pstRelay, IN TBUSADDR iPeerAddr,	IN TBUSSHMCHANNELCNF *a_pstChlCnf,
				  LPRELAYENV  a_pstRunEnv, IN LPTBUSSHMGCIMHEAD a_pstHead)
{
	int iRet = TBUS_SUCCESS ;
	RELAY *ptRelay = NULL;
	unsigned int i;

	assert(NULL != a_pstChlCnf);
	assert(NULL != a_pstRunEnv);
	assert(NULL != a_pstRelay);

	/*先检查此通道是否存在*/
	for (i = 0; i < a_pstRunEnv->dwRelayCnt; i++)
	{
		LPTBUSCHANNEL pstChl = &(a_pstRunEnv->pastTbusRelay[i]->stChl);
		if ((a_pstRelay->dwAddr == pstChl->pstHead->astAddr[pstChl->iRecvSide]) &&
			(iPeerAddr == pstChl->pstHead->astAddr[pstChl->iSendSide]))
		{
			TBUSD_RELAY_CLR_NOT_IN_CONF(a_pstRunEnv->pastTbusRelay[i]);
			TBUSD_RELAY_SET_ENABLE(a_pstRunEnv->pastTbusRelay[i]);
			break;
		}
	}
	if (i < a_pstRunEnv->dwRelayCnt)
	{
		return 0;
	}


	/*检查存储空间是否够*/
	if (a_pstRunEnv->dwRelayCnt >= sizeof(a_pstRunEnv->pastTbusRelay)/sizeof(a_pstRunEnv->pastTbusRelay[0]))
	{
		tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_ERROR,   "relay_channel_attach relay count reach max %d, cannot attach",
			a_pstRunEnv->dwRelayCnt) ;
		return -1;
	}


	/*为新的中转通道管理分配空间*/
	ptRelay = (RELAY *) malloc(sizeof(RELAY)) ;
	if (NULL == ptRelay)
	{
		tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_ERROR,   "relay_enable, failed to malloc memory<size:%d>", sizeof(RELAY)) ;
		return -2;
	}
	memset( ptRelay, 0, sizeof(RELAY) ) ;
	memcpy ( &(ptRelay->stRelayInfo), a_pstRelay, sizeof(TBUSSHMRELAYCNF) ) ;
	ptRelay->dwFlag = TBUSD_FLAG_ENABLE ;
	ptRelay->dwRelayStatus = RELAY_STATUS_NO_CONNECTION;
	ptRelay->iID = a_pstRunEnv->dwRelayCnt;

	iRet = tbus_attach_channel(a_pstRelay->dwAddr, a_pstChlCnf, &(ptRelay->stChl), a_pstHead) ;
	if (0 != iRet)
	{
		tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_ERROR,
			"failed to attach channel(0x%08x <--> 0x%08x), for %s", a_pstRelay->dwAddr,
			iPeerAddr, tbus_error_string(iRet)) ;
		free(ptRelay);
		ptRelay = NULL;
	}else
	{
		a_pstRunEnv->pastTbusRelay[a_pstRunEnv->dwRelayCnt] = ptRelay;
		a_pstRunEnv->dwRelayCnt++;
		tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_ERROR,
			"add channel(0x%08x <--> 0x%08x) to relaymgr,now relay channels num: %d", a_pstRelay->dwAddr,
			iPeerAddr, a_pstRunEnv->dwRelayCnt) ;
	}/*if (0 != iRet)*/

	return iRet ;

}

int tbusd_add_relay_channels(LPRELAYENV a_pstRunEnv, LPTBUSSHMGRM a_pstShmGRM, LPTBUSSHMGCIM a_pstShmGCIM)
{
	int iRet = 0;
	unsigned int	i = 0;
	unsigned int	j = 0;

	unsigned int dwRelaysCount;
	unsigned int dwChannelsCount;

	assert(NULL != a_pstRunEnv);
	assert(NULL != a_pstShmGRM);
	assert(NULL != a_pstShmGRM);

	dwRelaysCount = a_pstShmGRM->stHead.dwUsedCnt;
	dwChannelsCount = a_pstShmGCIM->stHead.dwUsedCnt;
	for ( i=0; i< dwRelaysCount; i++ )
	{
		LPTBUSSHMRELAYCNF pstRelay;

		pstRelay = &a_pstShmGRM->astRelays[i];
		if (!(pstRelay->dwFlag & TBUS_RELAY_ENABLE))
		{
			/* __log4c_category_trace ( g_ptRelayLog, "relay_channel_attach, relay info disable %i", i ) ; */
			continue ;
		}

		for ( j=0; j< dwChannelsCount; j++ )
		{
			TBUSADDR iPeerAddr;
			TBUSSHMCHANNELCNF *pstChannel = &(a_pstShmGCIM->astChannels[j]);

			if (!TBUS_GCIM_CHANNEL_IS_ENABLE(pstChannel))
			{
				continue;
			}

			if (pstRelay->dwAddr == pstChannel->astAddrs[0])
			{
				iPeerAddr = pstChannel->astAddrs[1];
			}else if (pstRelay->dwAddr == pstChannel->astAddrs[1])
			{
				iPeerAddr = pstChannel->astAddrs[0];
			}else
			{
				continue;
			}

			iRet = relay_enable(pstRelay, iPeerAddr, pstChannel, a_pstRunEnv,
				&a_pstShmGCIM->stHead);
			if ( TBUS_SUCCESS != iRet )
			{
				char szPeerAddr[32]={0};
				STRNCPY(szPeerAddr, tbus_addr_nota_by_addrtemplet(&a_pstShmGRM->stHead.stAddrTemplet, iPeerAddr),
					sizeof(szPeerAddr));
				tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_ERROR,   "failed to add channel(%s <-->%s) to ralay managei",
					szPeerAddr, tbus_addr_nota_by_addrtemplet(&a_pstShmGRM->stHead.stAddrTemplet, pstRelay->dwAddr)) ;
				break ;
			}
		}/*for ( j=0; j< dwChannelsCount; j++ )*/
		if (0 != iRet)
		{
			break;
		}
	}/*for ( i=0; i< dwRelaysCount; i++ )*/

	return iRet;
}

void tbusd_recycle_disabled_channels(LPRELAYENV a_pstRunEnv)
{
	time_t tNow;
	unsigned int i;

	assert(NULL != a_pstRunEnv);

	time(&tNow);
	for (i = 0; i < a_pstRunEnv->dwRelayCnt; i++)
	{
		RELAY *pstRelay = a_pstRunEnv->pastTbusRelay[i];
		CHANNELHEAD *pstHead = pstRelay->stChl.pstHead;

		if (TBUSD_RELAY_IS_ENABLE(pstRelay))
		{
			continue;
		}

		/*如果不在配置中的通道预留一段时间后超时，则关闭或尝试清除它*/
		if ((tNow - pstRelay->tBeginDisabled) >= TBUS_DISABLE_CHANNEL_CLEAR_DISABLE_TIMEOUTGAP)
		{
			tbusd_log_msg(g_ptRelayLog, TLOG_PRIORITY_FATAL, "the time which the channel(0x%08x <-->0x%08x) is disabled reach"
				"%d second, so clear the channel", pstHead->astAddr[0],
				pstHead->astAddr[1], TBUS_DISABLE_CHANNEL_CLEAR_DISABLE_TIMEOUTGAP);
			if (NULL != pstRelay->pConnNode)
			{
				/*关闭连接*/
				tbusd_destroy_conn(a_pstRunEnv, pstRelay->pConnNode);
				pstRelay->pConnNode = NULL;
			}
			free(pstRelay);
			if (i < a_pstRunEnv->dwRelayCnt -1 )
			{
				a_pstRunEnv->pastTbusRelay[i] = a_pstRunEnv->pastTbusRelay[a_pstRunEnv->dwRelayCnt -1];
				a_pstRunEnv->pastTbusRelay[a_pstRunEnv->dwRelayCnt -1] = NULL;
			}
			a_pstRunEnv->dwRelayCnt--;
			i--;
		}/*if ((tNow - pstChannel->tBeginDisable) >= TBUS_DISABLE_CHANNEL_CLEAR_DISABLE_TIMEOUTGAP)*/
	}/*for (i = 0; i < a_pstRunEnv->dwRelayCnt; i++)*/
}

int IsChannelNeedRelay(TBUSADDR	a_iAddr,  LPTBUSSHMGRM a_pstShmGRM)
{
	unsigned int i;

	assert(NULL != a_pstShmGRM);
	for (i =0 ; i < a_pstShmGRM->stHead.dwUsedCnt; i++)
	{
		if (a_iAddr == a_pstShmGRM->astRelays[i].dwAddr)
		{
			return 1;
		}
	}

	return 0;
}

int tbusd_refresh_relays_conf(LPRELAYENV a_pstRunEnv)
{
	int iRet = 0;
	int iNeedRefresh = 0;
	LPTBUSSHMGRM pstShmGRMBackup;
	LPTBUSSHMGRM pstShmGRM;
	LPTBUSSHMGCIM pstShmGCIM;
	unsigned int i;
	RELAY *pstRelay;
	CHANNELHEAD *pstChlHead ;
	time_t tNow;

	assert(NULL != a_pstRunEnv);


	/*首先检查回收不再使用的通道*/
	time(&tNow);
	tbusd_recycle_disabled_channels(a_pstRunEnv);


	/*检查gcim和grm的配置是否有更新*/
	pstShmGCIM = a_pstRunEnv->pstShmGCIM;
	assert(NULL != pstShmGCIM);
	if (a_pstRunEnv->dwGCIMVersion != pstShmGCIM->stHead.dwVersion)
	{
		a_pstRunEnv->bNeedRefresh = TDR_TRUE;
	}
	pstShmGRM = a_pstRunEnv->pstShmGRM;
	pstShmGRMBackup = &a_pstRunEnv->stShmGRMBackup;
	if (pstShmGRMBackup->stHead.dwVersion != pstShmGRM->stHead.dwVersion)
	{
		a_pstRunEnv->bNeedRefresh = TDR_TRUE;
	}
	if (a_pstRunEnv->bNeedRefresh == TDR_FALSE)
	{
		return 0; /*gcim和grm版本没有变化,因此不需要检查配置更新*/
	}

	/*先尝试获取grm配置的一个拷贝*/
	if (pstShmGRMBackup->stHead.dwVersion != pstShmGRM->stHead.dwVersion)
	{
		if (0 != pthread_rwlock_tryrdlock(&pstShmGRM->stHead.stRWLock))
		{
			return 0; /*锁定失败,下次在尝试*/
		}
		memcpy(&pstShmGRMBackup->stHead, &pstShmGRM->stHead, sizeof(pstShmGRMBackup->stHead));
		memcpy(&pstShmGRMBackup->astRelays[0], &pstShmGRM->astRelays[0],
			sizeof(TBUSSHMRELAYCNF)*(pstShmGRM->stHead.dwUsedCnt));
		pthread_rwlock_unlock(&pstShmGRM->stHead.stRWLock);
	}/*if (a_pstRunEnv->stShmGRMBackup.stHead.dwVersion != pstHead->dwVersion)*/



	/*尝试锁gcim配置，如果锁失败则下次再尝试*/
	if (0 != pthread_rwlock_trywrlock(&pstShmGCIM->stHead.stRWLock))
	{
		return 0;
	}

	/*先检查对于当前配置而言失效的中转通道*/
	for (i = 0; i < a_pstRunEnv->dwRelayCnt; i++)
	{
		pstRelay = a_pstRunEnv->pastTbusRelay[i];
		pstChlHead = pstRelay->stChl.pstHead;
		if (!IsChannelInGCIM(pstChlHead, pstShmGCIM))
		{
			/*relay中管理的通道不在gcim配置中，则此通道设置标志位*/
			tbusd_log_msg(g_ptRelayLog, TLOG_PRIORITY_DEBUG, "the relaychannel(0x%08x <--> 0x%08x)"
				"is not in gcim, so set the NOT_IN_CONF flag and disable it", pstChlHead->astAddr[pstRelay->stChl.iRecvSide],
				pstChlHead->astAddr[pstRelay->stChl.iSendSide]);
			TBUSD_RELAY_SET_NOT_IN_CONF(pstRelay);
			TBUSD_RELAY_CLR_ENABLE(pstRelay);
			pstRelay->tBeginDisabled = tNow;
			continue;
		}
		if (!IsChannelNeedRelay(pstChlHead->astAddr[pstRelay->stChl.iRecvSide],
			pstShmGRMBackup))
		{
			/*中转通道需要中转的一段不再需要中转，则此通道设置标志位*/
			tbusd_log_msg(g_ptRelayLog, TLOG_PRIORITY_DEBUG, "the relaychannel(0x%08x <--> 0x%08x)"
				"need not relay, so set the NOT_IN_CONF flag and disable it", pstChlHead->astAddr[pstRelay->stChl.iRecvSide],
				pstChlHead->astAddr[pstRelay->stChl.iSendSide]);
			TBUSD_RELAY_SET_NOT_IN_CONF(pstRelay);
			TBUSD_RELAY_CLR_ENABLE(pstRelay);
			pstRelay->tBeginDisabled = tNow;
			continue;
		}
	}/*for (i = 0; i < a_pstRunEnv->dwRelayCnt; i++)*/

	/*检查需要添加的通道*/
	iRet = tbusd_add_relay_channels(a_pstRunEnv, pstShmGRMBackup, pstShmGCIM);

	a_pstRunEnv->dwGCIMVersion = pstShmGCIM->stHead.dwVersion;

	/*解锁*/
	pthread_rwlock_unlock(&pstShmGCIM->stHead.stRWLock);

	a_pstRunEnv->bNeedRefresh = TDR_FALSE;

	return iRet;
}



int relay_channel_attach (LPRELAYENV  a_pstRunEnv)
{
	int iRet = TBUS_ERROR;
	LPTBUSSHMGCIM pstGCIM;
	LPTBUSSHMGRM pstGRM;


	assert(NULL != a_pstRunEnv);
	pstGCIM = a_pstRunEnv->pstShmGCIM;
	pstGRM = a_pstRunEnv->pstShmGRM;
	assert(NULL != pstGCIM);
	assert(NULL != pstGRM);


	/*1.GCIM和GRM通信地址模板是否一致,tbusd不做检查,其一致性由外部的配置管理系统保证
	  2.当机器上运行了多个tbusd时,如果不同时获取GCIM和GRM的锁,则,单独去锁GRM会导致死锁,这里采用一些折中
	的处理方法,每次只锁一个锁,将信息备份出来*/
	tbus_rdlock(&pstGRM->stHead.stRWLock);
	memcpy(&a_pstRunEnv->stShmGRMBackup.stHead, &pstGRM->stHead, sizeof(pstGRM->stHead));
	memcpy(&a_pstRunEnv->stShmGRMBackup.astRelays[0], &pstGRM->astRelays[0],
		sizeof(TBUSSHMRELAYCNF)*(pstGRM->stHead.dwUsedCnt));
	tbus_unlock(&pstGRM->stHead.stRWLock);
	pstGRM = &a_pstRunEnv->stShmGRMBackup;

	/*添加需要中转的通道*/
	tbus_wrlock(&pstGCIM->stHead.stRWLock);
	iRet = tbusd_add_relay_channels(a_pstRunEnv, pstGRM, pstGCIM);
	if (0 != iRet)
	{
		tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_ERROR, "failed to add relay channels, iRet:%d",
			iRet) ;
	}
	tbus_unlock(&pstGCIM->stHead.stRWLock);


	if (0 >= a_pstRunEnv->dwRelayCnt)
	{
		tlog_log(g_ptRelayLog, TBUSD_PRIOROTY_WARN,0,0,
			"relay_channel_attach, no  relay need to attach, please check configure") ;
	}


	return iRet ;
}





int app_epoll_create ( LPRELAYENV  a_pstRunEnv )
{
	int iRet = TBUS_SUCCESS ;

	assert(NULL != a_pstRunEnv);

	/* create epoll event pool */
	a_pstRunEnv->iEpoolFd = epoll_create ( RELAY_MAX_FD ) ;
	if ( 0 > a_pstRunEnv->iEpoolFd )
	{
		tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_ERROR,   "app_epoll_create, epoll_create() failed %s", strerror(errno)) ;
		iRet = TBUS_ERROR ;
	}


	tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_DEBU,   "RETN: app_epoll_create(void)=%i", iRet ) ;

	return iRet ;
}


int app_listen_create (LPRELAYENV  a_pstRunEnv)
{
	int iRet = TBUS_SUCCESS ;
	int s ;

	LPTBUSD pstConf;
	LPCONNNODE pstConn;

	assert(NULL != a_pstRunEnv);
	pstConf = a_pstRunEnv->pstConf;
	assert(NULL != pstConf);


	/* create listen handle */
	s = tnet_listen ( pstConf->szListen, pstConf->dwBackLog ) ;
	if ( 0 > s )
	{
		tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_ERROR,   "app_listen_create, tnet_listen() failed %i", s ) ;
		return TBUS_ERROR ;
	}


	/* register listen handle to epoll */
	pstConn = tbusd_create_conn(a_pstRunEnv, s, RELAY_FD_LISTEN_MODE, EPOLLIN);
	if (NULL == pstConn)
	{
		tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_ERROR,   "app_listen_create, failed to add fd %d to manage", s) ;
		return TBUS_ERROR;
	}

	tbusd_log_msg ( g_ptRelayLog,TBUSD_PRIORITY_DEBU,   "RETN: app_listen_create(), index %i, listen socket fd %i",
		pstConn->idx, s ) ;
	return iRet ;
}


int tbusd_proc(IN TAPPCTX *a_pstAppCtx,  IN LPRELAYENV a_pstRunEnv)
{
	int	iMaxSendPkg = TBUS_DEFAULT_SENT_CNT;
	LPTBUSDSTAT pstStat;
	struct timeval stStart;
	struct timeval stEnd;
	struct timeval stSub;
	int iRecvPkgNum = 0;
	int iSendPkgNum = 0;

	assert(NULL != a_pstAppCtx);
	assert(NULL != a_pstRunEnv);
	assert(NULL != a_pstRunEnv->pstStat);

	gettimeofday(&stStart, NULL);
	pstStat = a_pstRunEnv->pstStat;

	iRecvPkgNum = tbusd_proc_recv(a_pstAppCtx, a_pstRunEnv, &iMaxSendPkg);

	iSendPkgNum = tbusd_proc_send(a_pstRunEnv, iMaxSendPkg);


	gettimeofday(&stEnd, NULL);
	TV_DIFF(stSub, stEnd, stStart);
	if ((pstStat->iSecPreloop < (int)stSub.tv_sec))
	{
		pstStat->iSecPreloop = (int)stSub.tv_sec;
		pstStat->iUsecPreloop = (int)stSub.tv_usec;
	}else if (pstStat->iUsecPreloop < (int)stSub.tv_usec)
	{
		pstStat->iSecPreloop = (int)stSub.tv_sec;
		pstStat->iUsecPreloop = (int)stSub.tv_usec;
	}
	pstStat->dwLoopTimes++;

	return iRecvPkgNum + iSendPkgNum -1;
}



int tbusd_tick(TAPPCTX *a_pstAppCtx,  LPRELAYENV a_pstRunEnv)
{
#define RELAY_STAT_BUFF_SIZE  10240

	unsigned int i;
	LPTBUSD pstConf;


	assert(NULL != a_pstAppCtx);
	assert(NULL != a_pstRunEnv);
	pstConf = a_pstRunEnv->pstConf;
	assert(NULL != pstConf);

	/*write stat*/
	if (((a_pstAppCtx->stCurr.tv_sec - a_pstRunEnv->tLastStat) > pstConf->iStatGap) &&
		(0 < pstConf->iStatGap))
	{
		LPTBUSDSTAT pstStat = a_pstRunEnv->pstStat;
		for (i = 0; i < a_pstRunEnv->dwRelayCnt; i++)
		{
			LPTBUSRELAYCONNSTAT pstRelayStat = &pstStat->astRelayConnStat[i];
			pstStat->dwSendPkgCount += pstRelayStat->dwSendPkgCount;
			pstStat->ullSendPkgBytes += pstRelayStat->ullSendPkgBytes;
			pstStat->dwSendSynPkgCount += pstRelayStat->dwSendSynPkgCount;
			pstStat->dwRecvSynPkgCount += pstRelayStat->dwRecvSynPkgCount;
		}
		tbusd_log_data(g_ptStatLog, TBUSD_PRIORITY_FATAL, (LPTDRMETA)a_pstAppCtx->stRunDataStatus.iMeta,
			a_pstAppCtx->stRunDataStatus.pszBuff, a_pstAppCtx->stRunDataStatus.iLen);


		tbusd_init_stat(a_pstRunEnv);
		a_pstRunEnv->tLastStat = a_pstAppCtx->stCurr.tv_sec;
	}/*if ((0 < pstConf->iStatGap) && (a_pstAppCtx->stCurr.tv_sec - sLastStat > pstConf->iStatGap))*/



	/*定时检查gcim和grm的配置,刷新通道配置*/
	if ((0 < a_pstAppCtx->stOption.iRefeshTbusTimer) &&
		((a_pstAppCtx->stCurr.tv_sec - a_pstRunEnv->tLastRefreshConf) >= a_pstAppCtx->stOption.iRefeshTbusTimer))
	{
		tbusd_refresh_relays_conf(a_pstRunEnv);
		a_pstRunEnv->tLastRefreshConf = a_pstAppCtx->stCurr.tv_sec;
	}

	/*对于尚未建立握手过程的relay通道，重建握手过程*/
	tbusd_conncet_peer(a_pstRunEnv, a_pstAppCtx->stCurr.tv_sec) ;

	tbusd_calc_pkglimit_persend(a_pstRunEnv);


	/*对于握手超时的relay链接，主动断开链接*/
	if (0 < pstConf->iHandShakeTimeout)
	{
		tbusd_check_handshake(a_pstRunEnv, a_pstAppCtx->stCurr.tv_sec);
	}


	/*定时同步序列号*/
	if ((0 < pstConf->iSynSeqGap) && (0 != pstConf->iNeedAckPkg))
	{
		tbusd_tick_synseq(a_pstRunEnv, a_pstAppCtx->stCurr.tv_sec);
	}



	return TBUS_SUCCESS;
}

/** 当tbusd idle时将epoll_wait的时间改成 10毫秒等待
*/
int tbusd_idle(IN TAPPCTX *a_pstAppCtx,  IN LPRELAYENV a_pstRunEnv)
{
	assert(NULL != a_pstAppCtx);
	assert(NULL != a_pstRunEnv);

	a_pstRunEnv->iEpollWaitTime = a_pstAppCtx->iEpollWait;

	return 0;
}


